package com.hotutil.start;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.DFSInotifyEventInputStream;
import org.apache.hadoop.hdfs.client.HdfsAdmin;
import org.apache.hadoop.hdfs.inotify.Event;
import org.apache.hadoop.hdfs.inotify.EventBatch;
import org.apache.hadoop.hdfs.inotify.MissingEventsException;
import org.apache.log4j.Logger;


import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;


public class HDFS_HotUpdate {
    //这里用static声明两个变量，因为static声明后的元素被称为gc roots， GC不会主动回收gc roots的变量。用static避免内存泄露
    public static HdfsAdmin admin =null;
    public static DFSInotifyEventInputStream inotifyEventStream=null;
    public static Configuration configuration = new Configuration();
    public static FileSystem fs =null;
    public static EsUtil esUtil=new EsUtil("10.7.221.29",9200);
    public static String indexName="officefile";
    
    //日志记录器
    /**
     * 级别
     * logger.fatal("致命问题");
     * logger.error("报错");
     * logger.warn("这是警告");
     * logger.info("控制台打印信息");
     * logger.debug("这是debug!!!!");
     */
    public static Logger logger=Logger.getLogger(HDFS_HotUpdate.class);

    /**
     * 解析edit文件信息的方法，通过此方法监控Test02目录
     */
    public static void main(String[] args)  {
        for (int i=0;i<10;i++) {
            try {
                TestEvent();
                i=0;
            } catch (Exception e) {
                logger.error(e.getMessage());
                if (i == 9) {
                    logger.error("---------程序重启失败,请排查HDFS和ElasticSearch后重新运行程序---------");
                    System.exit(0);
                }
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
        }




    }
    public static void TestEvent() throws IOException, MissingEventsException, InterruptedException, URISyntaxException {
        /*
            先设置系统环境变量，这里的"系统"指的是JVM,不是OS.
            因为Hadoop登录时的login()方法，会调用HadoopLoginModule的login()方法和commit()方法,HadoopLoginModule的login
            方法只会打印一行调试日志。commit()方法在使用Kerberos的时候，会优先从Kerberos去获取username,在未使用kerberos的情况下，
            读取系统HADOOP_USER_NAME系统环境变量，如果不为空拿它作为username,否则，读取HADOOP_USER_NAME这个JVM变量。否则从否则，
            从com.sun.security.auth.NTUserPrincipal或者com.sun.security.auth.UnixPrincipal的实例获取username。如果以上尝试都失败，那么抛出异常LoginException("Can’t find user name")。
            最终拿username构造org.apache.hadoop.security.User的实例添加到Subject中。
             参考HADOOP登录过程 http://www.udpwork.com/item/7047.html
        */
        System.setProperty("HADOOP_USER_NAME","hdfs");
        //下面两个配置在生产环境再打开
        configuration.addResource(new Path("/etc/hadoop/3.1.4.0-315/0/core-site.xml"));
        configuration.addResource(new Path("/etc/hadoop/3.1.4.0-315/0/hdfs-site.xml"));
        URI uri=new URI("hdfs://10.7.221.29:8020");
        admin = new HdfsAdmin(uri,configuration);
        fs = FileSystem.get(uri, configuration);
        FSDataInputStream fis=null;

        /*
            getInotifyEventStream（），获取事件流的对象
            注意：此方法必须是单线程,不能在多个线程之间共享流对象。
            获取流的方式：当一个文件被DN读取后，会生成一个MetadataUpdateEvent对象并发送到NN，如果发生修改（追加、删除、移动、复制），attime会更新，再次发送MetadataUpdateEvent
            不需要我们手动获取，这里会自动接收DN推送的Event.
         */
        inotifyEventStream = admin.getInotifyEventStream();

        while (true){


            /*
                手动获取一批新的事件，如果不能立即获取到，则会无限期的卡在这里，一直等待,该功能也是避免死循环导致内存泄露的一种手段。
                take()方法官方解释原文：Returns the next batch of events in the stream, waiting indefinitely if
   * a new batch  is not immediately available.
             */
            EventBatch events = inotifyEventStream.take();
            for (Event event: events.getEvents()){
                //事件类型
//                System.out.println("事件类型是：" + event.getEventType());
                switch (event.getEventType()){
                    //通过判断文件夹
                    //一是创建文件（上传）；二是删除文件；三是更改文件（夹）；四是移动文件（夹）
                    //删除文件夹，只提示删除文件夹，不显示删除文件
                    //WebHDFS的删除不会进回收站
                    case RENAME:
                        Event.RenameEvent rename = (Event.RenameEvent) event;
                        //源路径
                        String srcPath = rename.getSrcPath();
                        //目标路径
                        String dstPath = rename.getDstPath();
                        //如果源地址是以目的地址开头的，则可以判断是临时文件。当其rename之后，就变成了最终文件
                        if (srcPath.startsWith(dstPath)&&srcPath.endsWith("_COPYING_")&&dstPath.startsWith("/")&&(dstPath.toLowerCase().endsWith(".doc")||dstPath.toLowerCase().endsWith(".docx")||dstPath.toLowerCase().endsWith(".pdf")||dstPath.toLowerCase().endsWith(".xlsx")||dstPath.toLowerCase().endsWith("xls")||dstPath.toLowerCase().endsWith("txt")||dstPath.toLowerCase().endsWith("csv"))){
                            try {
                                //1.通过open方法获取一个流
                                fis = fs.open(new Path(dstPath));
                                //2.通过TikaUtil解析这个流的内容
                                String result = TikaUtil.tikaParse(fis);
                                //3.将解析到的内容，写到ES里
                                String officefile = esUtil.createDoc(indexName, dstPath, dstPath, result);
                                if (officefile.equals("created")){
                                    logger.info("文档------"+dstPath+"------已添加到ElasticSearch");
                                }else{
                                    logger.info("文档------"+dstPath+"------添加ES时失败，请检查ES");
                                }
                            }catch (IOException ex)
                            {
                                logger.error("文档------"+dstPath+"------添加至es时失败，错误原因:------"+ex.getMessage());
                            }finally {
                                fis.close();
                            }


                        }
                        /**如果源地址的路径是以过滤路径开头，目的地址是以垃圾桶路径开头。则认为是删除目录（不跳过垃圾桶）
                         else if (dstPath.startsWith("/user")&srcPath.startsWith("/test02")){
                         System.out.println("正在使用非restful风格删除文件，路径为：" + srcPath);
                         }
                         **/
                        //剩下的情况就是修改文件（夹）名称、或者移动文件（夹）
                        else {
                            if (dstPath.startsWith("/")&&srcPath.startsWith("/")&&(dstPath.toLowerCase().endsWith(".doc")||dstPath.toLowerCase().endsWith(".docx")||dstPath.toLowerCase().endsWith(".pdf")||dstPath.toLowerCase().endsWith(".xlsx")||dstPath.toLowerCase().endsWith("xls")||dstPath.toLowerCase().endsWith("txt")||dstPath.toLowerCase().endsWith("csv"))){
                                if (dstPath.split("/").length == srcPath.split("/").length) {
                                    //修改名称,修改名称后，先把原来的文档删除掉，再改成新的
                                    String del = esUtil.deleteDoc(indexName, srcPath);

                                    if(del.equals("deleted")){
                                        try {
                                            //1.通过open方法获取一个流
                                            fis = fs.open(new Path(dstPath));
                                            //2.通过TikaUtil解析这个流的内容
                                            String result = TikaUtil.tikaParse(fis);
                                            //3.将解析到的内容，写到ES里
                                            String officefile = esUtil.createDoc(indexName, dstPath, dstPath, result);
                                            logger.info("文档------"+srcPath+"------已更名，新名称为------"+dstPath+"------");
                                        }catch (Exception ex){
                                            logger.error("文档------"+srcPath+"------已更名,但更新时索引损坏，请根据新路径重新创建------"+dstPath+"------");
                                        }finally {
                                            fis.close();
                                        }
                                    }else{
                                        logger.error("ES删除索引------"+srcPath+"------时失败，这个错误会导致索引无法更新");
                                    }
                                }else {
                                    //移动文件时，先删除源文件索引，再根据新文件位置创建索引
                                    String del = esUtil.deleteDoc(indexName, srcPath);

                                    if(del.equals("deleted")){
                                        try {
                                            //1.通过open方法获取一个流
                                            fis = fs.open(new Path(dstPath));
                                            //2.通过TikaUtil解析这个流的内容
                                            String result = TikaUtil.tikaParse(fis);
                                            //3.将解析到的内容，写到ES里
                                            String officefile = esUtil.createDoc(indexName, dstPath, dstPath, result);
                                            logger.info("文档------"+srcPath+"------已移动,新路径为------"+dstPath+"------");
                                        }catch (Exception ex){
                                            logger.error("文档------"+srcPath+"------已移动,但更新时索引损坏，请根据新路径重新创建------"+dstPath+"------");
                                        }finally {
                                            fis.close();
                                        }
                                    }else{
                                        logger.error("ES删除索引------"+srcPath+"------时失败，这个错误会导致索引无法更新");
                                    }

                                }
                            }
                        }
                        break;

                    //3、未知操作：比如跳过回收站删除文件（夹）
                    case UNLINK:
                        Event.UnlinkEvent unlinkEvent = (Event.UnlinkEvent) event;
                        String unlinkPath = unlinkEvent.getPath();
                        if (unlinkPath.startsWith("/")&&(unlinkPath.endsWith(".doc")||unlinkPath.endsWith(".docx")||unlinkPath.endsWith(".pdf")||unlinkPath.endsWith(".xlsx")||unlinkPath.endsWith("xls")||unlinkPath.endsWith("txt")||unlinkPath.endsWith("csv"))){
                            String del = esUtil.deleteDoc(indexName,unlinkPath);
                            if(del.equals("deleted")) {
                                logger.info("索引" + del + "已删除");
                            }else{
                                logger.error("删除索引------"+unlinkPath+"------时失败,这个错误不会导致任何情况发生");
                            }
                        }
                        break;
                    default:
                        break;
                }
            }
        }
    }


}
